package com.project.modules.netty;

import cn.hutool.core.collection.ConcurrentHashSet;
import io.netty.channel.Channel;
import io.netty.channel.ChannelHandler;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.handler.codec.http.websocketx.TextWebSocketFrame;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component;

import java.util.concurrent.ConcurrentHashMap;

/**
 * 自定义处理类
 */
@Component
@ChannelHandler.Sharable
@Slf4j
public class WebSocketHandler extends SimpleChannelInboundHandler<TextWebSocketFrame> {

    /**
     * 用户通道管理
     * key：通道 id
     * value：通道对象
     */
    public static ConcurrentHashMap<String, Channel> channelMap = new ConcurrentHashMap<>(16);

    /**
     * 红包频道管理
     * key: 红包 id
     * value： 通道 id 的 set
     */
    public static ConcurrentHashMap<Long, ConcurrentHashSet<String>> redChannels = new ConcurrentHashMap<>();

    /**
     * 发送消息
     *
     * @param redId           红包 id
     * @param responseMessage 消息内容
     */
    public static void sendMessage(Long redId, String responseMessage) {
        for (String channel : redChannels.getOrDefault(redId, new ConcurrentHashSet<>())) {
            Channel c = channelMap.get(channel);
            if (c != null && c.isActive() && c.isWritable()) {
                c.writeAndFlush(new TextWebSocketFrame(responseMessage));
                log.info("发送消息：" + responseMessage);
            }
        }
    }

    /**
     * 收到消息事件
     *
     * @param ctx 通道处理程序上下文
     * @param msg 文本框架网络套接字
     */
    @Override
    protected void channelRead0(ChannelHandlerContext ctx, TextWebSocketFrame msg) {
        log.info("msg:" + msg.text());
        String[] params = msg.text().split("###");
        if (params.length < 2) {
            log.error("消息格式错误" + msg.text());
            return;
        }
        String channelId = ctx.channel().id().asLongText();
        Long redId = Long.valueOf(params[0]);
        // 更新红包频道中的通道
        ConcurrentHashSet<String> channelIds = redChannels.getOrDefault(redId, new ConcurrentHashSet<>());
        channelIds.add(channelId);
        redChannels.put(redId, channelIds);

        String userName = params[1];
        if (params.length == 2) {
            // xxx来了
            sendMessage(redId, "@welcome###" + userName);
            return;
        }
        String message = params[2];
        if (message.startsWith("@")) {
            // 指令消息
            sendMessage(redId, message + "###" + userName);
            return;
        }
        // 群发聊天消息
        sendMessage(redId, userName + "###" + message);
    }

    /**
     * 通道就绪事件
     */
    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        Channel channel = ctx.channel();
        // 放入通道组
        channelMap.put(channel.id().asLongText(), channel);
        channel.writeAndFlush(new TextWebSocketFrame(channel + "上线了"));
        log.info("[" + channelMap.size() + "]有新的连接:" + channel.remoteAddress());
    }

    /**
     * 通道未就绪事件
     */
    @Override
    public void channelInactive(ChannelHandlerContext ctx) throws Exception {
        Channel channel = ctx.channel();
        log.info("通道：" + channel.remoteAddress() + "已下线");
        // 当有客户端断开连接的时候,就移除对应的通道
        channelMap.remove(channel.id().asLongText());
        ctx.close();
    }

    /**
     * 异常处理事件
     *
     * @param ctx   ctx
     * @param cause 导致
     * @throws Exception 异常
     */
    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        Channel channel = ctx.channel();
        //移除集合
        channelMap.remove(channel.id().asLongText());
        ctx.close();
        log.error("异常信息：{}", cause.getMessage());
    }
}

